package com.example.wxg.batch;

import com.example.wxg.stream.TableApiTest;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.java.BatchTableEnvironment;
import org.apache.flink.table.sinks.CsvTableSink;


/**
 * @author void
 * @date 2021/11/12 15:52
 * @desc
 */
public class SqlTest {

    public static void main(String[] args) throws Exception {
        ExecutionEnvironment bEnv = ExecutionEnvironment.getExecutionEnvironment();
        BatchTableEnvironment bTableEnv = TableEnvironment.getTableEnvironment(bEnv);

        DataSource<String> dataSource = bEnv.readTextFile("D://data/student.txt");
        DataSet<TableApiTest.Student> inputData = dataSource.map(new MapFunction<String, TableApiTest.Student>() {
            @Override
            public TableApiTest.Student map(String s) throws Exception {
                String[] splits = s.split(",");
                return new TableApiTest.Student(splits[0], Integer.parseInt(splits[1]));
            }
        });
        
        //DataSet转化为Table
        Table table = bTableEnv.fromDataSet(inputData);
        //注册表
        bTableEnv.registerTable("student", table);
        Table sqlQuery = bTableEnv.sqlQuery("select count(1),avg(age) from student");
        
        //创建sink
        CsvTableSink csvTableSink = new CsvTableSink("D://data/result.csv", ",", 1, FileSystem.WriteMode.OVERWRITE);
        //注册sink
        bTableEnv.registerTableSink("csvOutputTable", new String[]{"count", "avg_age"}, new TypeInformation[]{Types.LONG, Types.INT}, csvTableSink);
        sqlQuery.insertInto("csvOutputTable");
        bEnv.execute("SQL批次");
    }
}
